package com.jeesite.modules.isys.xmqtt.client;



import com.jeesite.modules.isys.xmqtt.config.XMqttConfig;
import com.jeesite.modules.isys.xmqtt.entity.QosEnum;
import lombok.Data;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


import java.util.Set;

@Data
public class XmqttClient{
    private IMqttClient mqttClient;
    private XMqttCallback xMqttCallback;
    private XMqttConfig xmqttConfig;
    private MqttClientPersistence memoryPersistence;

    public void restart() throws MqttException {
        mqttClient = new MqttClient(xmqttConfig.getBroker(),xmqttConfig.getClientId(),memoryPersistence);

        this.connect(xmqttConfig.getUsername(),xmqttConfig.getPassword());
        //获取所有订阅
        Set<String> topics = XSubscriptions.getTopics();
        for (String topic : topics){
            mqttClient.subscribe(topic);
        }
    }

    /**
     * 初始化配置
     *
     */
    public XmqttClient(XMqttConfig xmqttConfig, XMqttCallback xMqttCallback) {
         memoryPersistence = new MemoryPersistence();
        this.xMqttCallback = xMqttCallback;
        this.xmqttConfig = xmqttConfig;
        try {
            mqttClient = new MqttClient(xmqttConfig.getBroker(),xmqttConfig.getClientId(),memoryPersistence);

            this.connect(xmqttConfig.getUsername(),xmqttConfig.getPassword());
            //获取所有订阅
            Set<String> topics = XSubscriptions.getTopics();
            for (String topic : topics){
                mqttClient.subscribe(topic);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 连接broker
     * @param username
     * @param password
     */
    public void connect(String username,String password) {
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        connectOptions.setAutomaticReconnect(true);
        connectOptions.setCleanSession(true);
        connectOptions.setUserName(username);
        connectOptions.setPassword(password.toCharArray());
        mqttClient.setCallback(xMqttCallback);
        try {
            mqttClient.connect(connectOptions);
        } catch (MqttException e) {
            e.printStackTrace();
        }

    }


    /***
     * 发布
     * @param topic
     * @param msg
     * @param qos
     * @param retain
     */
    public void publish(String topic, Object msg, QosEnum qos, boolean retain) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos.value());
        mqttMessage.setRetained(retain);
        mqttMessage.setPayload(String.valueOf(msg).getBytes());
        if (mqttClient.isConnected()) {
            try {
                mqttClient.publish(topic,mqttMessage);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

    /***
     * 发布
     * @param topic
     * @param msg
     * @param qos
     * @param retain
     */
    public void publish(String topic, MqttMessage msg, QosEnum qos, boolean retain) {
        if (mqttClient.isConnected()) {
            try {
                mqttClient.publish(topic,msg);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }


    /***
     * 订阅
     * @param topicFilter
     * @param qos
     */
    public void subscribe(String topicFilter,QosEnum qos) {
        try {
            mqttClient.subscribe(topicFilter,qos.value());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /***
     * 断开连接
     */
    public void disConnect() {
        try {
            mqttClient.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}
